package com.acmedcare.framework.exchange.center;

import com.acmedcare.framework.exchange.center.consumer.ConsumerCommandListener;
import com.acmedcare.framework.exchange.center.consumer.IConsumerProxyContainer;
import com.acmedcare.framework.exchange.center.core.CommonCommandListener;
import com.acmedcare.framework.exchange.center.core.Const;
import com.acmedcare.framework.exchange.center.core.PluginContainerMBean;
import com.acmedcare.framework.exchange.center.core.handler.StageCommandHandler;
import com.acmedcare.framework.exchange.center.core.storage.ConsumerRepository;
import com.acmedcare.framework.exchange.center.core.storage.PluginRepository;
import com.acmedcare.framework.exchange.center.entity.bean.Consumer;
import com.acmedcare.framework.exchange.center.entity.bean.Plugin;
import com.acmedcare.framework.exchange.center.entity.dto.Hdfs;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.locks.ReentrantLock;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.stereotype.Service;

@Service
public class ApplicationReadyEventListener {

  final PluginRepository repository;

  final Hdfs hdfs;

  final PluginContainerMBean pluginContainer;

  final IConsumerProxyContainer consumerProxyContainer;

  final ConsumerRepository consumerRepository;
  final RabbitTemplate rabbitTemplate;

  final ConsumerCommandListener consumerCommandListener;

  final StageCommandHandler stageCommandHandler;

  final ReentrantLock reentrantLock = new ReentrantLock();
  static Boolean isLoaded = false;

  public ApplicationReadyEventListener(
      PluginRepository repository, Hdfs hdfs,
      PluginContainerMBean pluginContainer,
      IConsumerProxyContainer consumerProxyContainer,
      ConsumerRepository consumerRepository,
      RabbitTemplate rabbitTemplate,
      ConsumerCommandListener consumerCommandListener,
      StageCommandHandler stageCommandHandler) {

    this.repository = repository;
    this.hdfs = hdfs;
    this.pluginContainer = pluginContainer;
    this.consumerProxyContainer = consumerProxyContainer;
    this.consumerRepository = consumerRepository;
    this.rabbitTemplate = rabbitTemplate;
    this.consumerCommandListener = consumerCommandListener;
    this.stageCommandHandler = stageCommandHandler;
  }

  public void init() {
    reentrantLock.lock();
    if (isLoaded) {

      reentrantLock.unlock();
      return;
    }

    isLoaded = true;
    reentrantLock.unlock();
    addStageListener();
    List<Plugin> plugins = repository.getPlugins();
    pluginContainer.init(plugins);
    initConsumer();
  }

  private void initConsumer() {

    List<Consumer> consumers = consumerRepository.getConsumers();

    if (consumers != null) {
      consumerProxyContainer.init(consumers);
    }
    addConsumerListener(consumerCommandListener);
  }

  private SimpleMessageListenerContainer addConsumerListener(
      ConsumerCommandListener consumerCommandListener) {

    return CommonCommandListener
        .addFanoutListener(new MessageListenerAdapter(consumerCommandListener, "receive"),
            Const.ConsumerExchangeName,
            "fanout.exchange.center.open.api.consumer." + System.getenv(Const.GroupName) + "."
                + UUID
                .randomUUID().toString(), rabbitTemplate);
  }

  private void addStageListener() {

    stageCommandHandler.loadConfig();
    MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(stageCommandHandler,
        "receive");
    CommonCommandListener
        .addFanoutListener(messageListenerAdapter,
            Const.StageExchangeName,
            "fanout.exchange.center.open.api.stage." + System.getenv(Const.GroupName) + "."
                + UUID.randomUUID()
                .toString(), rabbitTemplate);
  }
}
